package com.sunlands.zlcx.usercenter.service.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.sunlands.zlcx.usercenter.client.DeskMatePushService;
import com.sunlands.zlcx.usercenter.client.SocketService;
import com.sunlands.zlcx.usercenter.client.bean.PushDTO;
import com.sunlands.zlcx.usercenter.client.bean.UserOnlineVO;
import com.sunlands.zlcx.usercenter.common.annotation.AutoLog;
import com.sunlands.zlcx.usercenter.common.cache.RedisDistributedLockImpl;
import com.sunlands.zlcx.usercenter.constant.RedisKeyConst;
import com.sunlands.zlcx.usercenter.constant.YN;
import com.sunlands.zlcx.usercenter.domain.UsersDO;
import com.sunlands.zlcx.usercenter.exception.CheckRunException;
import com.sunlands.zlcx.usercenter.exception.SendExceptionUtil;
import com.sunlands.zlcx.usercenter.service.UserRecommendServiceImpl;
import com.sunlands.zlcx.usercenter.service.UsersServiceImpl;
import com.sunlands.zlcx.usercenter.service.UsersStudyServiceImpl;
import com.sunlands.zlcx.usercenter.util.AsyncTaskManager;
import com.sunlands.zlcx.usercenter.util.StringUtil;
import com.sunlands.zlcx.usercenter.vo.LeaveMsgVO;
import com.sunlands.zlcx.usercenter.vo.UsersVO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.JedisCommands;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.sunlands.zlcx.usercenter.config.RedisConfig.CacheKeys.*;
import static com.sunlands.zlcx.usercenter.constant.LearningStatus.LEARN_TIMING;

@Service
@Slf4j
@EnableScheduling
public class TaskServiceImpl {

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    AsyncTaskManager asyncTaskManager;

    @Autowired
    RedisDistributedLockImpl redisDistributedLock;

    @Autowired
    private DeskMatePushService deskMatePushService;

    @Autowired
    private UsersStudyServiceImpl usersStudyService;

    @Autowired
    private UsersServiceImpl usersService;

    @Autowired
    private SocketService socketService;

    @Resource(name = "gyfTaskExecutor")
    private AsyncTaskExecutor gyfTaskExecutor;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private UserRecommendServiceImpl userRecommendService;

    @Value("${eureka.debug:false}")
    private Boolean debug;

    @Value("${limitUserUrl}")
    private String limitUserUrl;

    @Autowired
    private SendExceptionUtil sendExceptionUtil;

//    @Scheduled(cron = "${scheduled.timepush}")
//    public void newUsePush() {
////
////        if (debug || sendExceptionUtil.getSysdev().toLowerCase().equals("pre")) {
////            log.warn("本地环境不启动定时");
////            return;
////        }
//
//        String taskKey = RedisKeyConst.getTaskKey();
//        if (StringUtil.isEmpty(taskKey)) {
//            return;
//        }
//        if (!redisDistributedLock.lock(taskKey, 10000, 10000, 500, 1)) {
//            return;
//        }
//        asyncTaskManager.getExecutor().submit(() -> {
//            try {
//                String zSortedKey = RedisKeyConst.getUserSetLearnTime();
//                Set<ZSetOperations.TypedTuple<Integer>> pushKeySet = redisTemplate.opsForZSet().rangeWithScores(zSortedKey, 0, -1);
//                if (CollectionUtils.isEmpty(pushKeySet)) {
//                    return;
//                }
//                log.debug("设定时间的key push key = {}", JSON.toJSONString(pushKeySet));
//                for (ZSetOperations.TypedTuple<Integer> pushKey : pushKeySet) {
//                    Integer userId = pushKey.getValue();
//                    try {
//                        JSONObject jsonObject1 = usersStudyService.getLearnTimeByUserId(userId.longValue());
//                        long learnTimeByUserId = jsonObject1.getLong("learnTime");
//                        log.debug("learnTimeByUserId ={}, v={}, s={}", learnTimeByUserId, pushKey.getValue(), pushKey.getScore());
//                        if (learnTimeByUserId >= pushKey.getScore()) {
//                            Long remove = redisTemplate.opsForZSet().remove(zSortedKey, userId);
//                            log.debug("删除定时key zSortedKey={}, userId={}, remove={}", zSortedKey, userId, remove);
//                            PushDTO pushDTO = PushDTO.builder().content("您预订的计时时间已到").type(1).ids(Lists.newArrayList(userId + "")).build();
//                            JSONObject jsonObject = deskMatePushService.push(pushDTO);
//                            if (null == jsonObject || null == jsonObject.getInteger("code") || 0 != jsonObject.getInteger("code")) {
//                                throw new CheckRunException("push系统异常");
//                            }
//                        }
//                    } catch (Exception ex) {
//                        log.error("", ex);
//                    }
//                }
//            } catch (Exception e) {
//                log.error("", e);
//            }
//
//        });
//
//        redisDistributedLock.release(taskKey);
//
//    }

    public static String UPDATE_HOMEPAGE_USER_LOCK = "tz-user-center:update-homepage-users";
    /**
     * 定时刷新用户池缓存 凌晨3点15刷新
     */
    @AutoLog
    @Scheduled(cron = "0 15 3 * * ?")
    public void scheduleFlushCache() {
        if (debug) {
            log.warn("本地环境不启动定时");
            return;
        }
        //抢占锁，只允许一台机器执行
        RLock lock = redissonClient.getLock(UPDATE_HOMEPAGE_USER_LOCK);
        try {
            if (lock.tryLock(1, 5, TimeUnit.SECONDS)) {
                log.info("开始刷新用户池");
                long t = System.currentTimeMillis();
                userRecommendService.flushUserPool();
                long t1 = System.currentTimeMillis();
                log.info("刷新用户池结束 耗时{}", t1 - t);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


//    @AutoLog
//    @Scheduled(cron = "0 0/1 * * * ?")
//    public void limitLogin(){
//        DicDO dicDO = dicService.findByCode(LIMIT_LOGIN_PHONE.dicName());
//
//        List<String> list = Arrays.asList(dicDO.filter(device_ids.name()).getData().split(","));
//        if(list.isEmpty()){
//            return;
//        }
//
//        List<UsersDO> users = usersService.findByLimitDeviceUser(list);
//        if(users.isEmpty()){
//            return;
//        }
//
//        log.info("查询到被封禁设备的账号 users = {}, 进行封号处理！", users);
//        StringBuilder userString = new StringBuilder("查询到被封禁设备的账号，进行封号处理 \n");
//        users.forEach(u -> {
//            String result = HttpUtil.doGet(limitUserUrl + u.getId());
//            userString.append("\nid = ");
//            userString.append(u.getId());
//            userString.append("\n");
//            userString.append("name = ");
//            userString.append(u.getName());
//            userString.append("\n");
//            userString.append("deviceId = ");
//            userString.append(u.getDeviceId());
//            userString.append(" \n结果:");
//            userString.append(result);
//        });
//
//        if(!users.isEmpty()){
//            sendExceptionUtil.sendException(userString.toString());
//        }
//    }

    /**
     * 巡检用户信息表中用户通道和在线状态并进行检查更正
     */
    @AutoLog
    @Scheduled(cron = "0 0/10 * * * ?")
    public void refreshUserOnlineState() {
        if (debug) {
            log.warn("本地环境不启动定时");
            return;
        }
        if (debug || sendExceptionUtil.getSysdev().toLowerCase().equals("pre")) {
            log.warn("本地环境不启动定时");
            return;
        }

        try {
            if (!getRefreshUserOnlineLock()) {
                return;
            }

            //双向检查
            //查询数据库中所有在线用户状态是否和通道中一致
            int i = 0;
            Pageable pageable = new PageRequest(i, 1000);
            Page<UsersDO> page = null;
            while (Objects.isNull(page) || page.hasNext()) {
                page = usersService.findOnlineUsers(pageable);
                List<UserOnlineVO> allChannel = socketService.getOnline(null).getData();
                for (UsersDO user : page.getContent()) {
                    UserOnlineVO online = allChannel.stream().filter(u -> u.getUserId().equals(user.getId())).findFirst().orElse(null);
                    if (Objects.nonNull(online)) {
                        if ((Objects.nonNull(online.getTopRun()) && !online.getTopRun().equals(user.getTopRun()))
                                || (Objects.nonNull(online.getChannelId()) && !online.getChannelId().equals(user.getChannelId()))) {
                            Integer index = usersService.updateOnlineById(user.getId(), YN.YES.getCode(), online.getTopRun(), online.getChannelId(), online.getBackRunTime());
                            log.info("用户在线状态检查，用户在线，前后台状态不一致，进行修正 结果：index = {}  userId = {} online = {} local-topRun = {} local-channelId = {}", index, user.getId(), online, user.getTopRun(), user.getChannelId());
                        }
                        log.debug("用户在线状态检查，数据一致，不进行修改 userId = {}", user.getId());
                    } else {
                        Integer index = usersService.updateOnlineById(user.getId(), YN.NO.getCode(), null, user.getChannelId(), 0L);
                        log.info("用户在线状态检查，用户不在线，进行修正 结果：index = {} userId = {}  online = {} local-topRun = {} local-channelId = {}", index, user.getId(), online, user.getTopRun(), user.getChannelId());
                    }
                    i++;
                }
                Thread.sleep(500);
            }

            //检查通道的状态和数据库中是否一致
            List<UserOnlineVO> allChannel = socketService.getOnline(null).getData();
            if(allChannel.isEmpty()){
                return;
            }
            List<Long> userOnline = allChannel.stream().map(UserOnlineVO::getUserId).collect(Collectors.toList());
            List<List<Long>> partition = Lists.partition(userOnline, 1000);
            partition.forEach(userIdList -> {
                if(userIdList.isEmpty()){
                    return;
                }
                log.info("userIdList = {}", userIdList);
                List<UsersDO> offLineUsers = usersService.findOffLineUsers(userIdList);
                log.debug("offLineUsers = {}", offLineUsers);
                offLineUsers.forEach(user -> {
                    UserOnlineVO channelInfo = allChannel.stream().filter(online -> online.getUserId().equals(user.getId())).findFirst().orElseThrow(() -> new NullPointerException("没有在通道集合中找到用户 userId = " + user.getId()));
                    Integer index = usersService.updateOnlineById(user.getId(), YN.YES.getCode(), channelInfo.getTopRun(), channelInfo.getChannelId(), channelInfo.getBackRunTime());
                    log.info("用户在线状态检查，用户在线，数据库状态显示不在线，进行修正 结果：index = {}  userId = {} online = {} local-topRun = {} local-channelId = {}", index, user.getId(), channelInfo, user.getTopRun(), user.getChannelId());
                });
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });


        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            releaseRefreshUserOnlineLock();
        }
    }


    private static final long leaveMinute = 60;


    @AutoLog
    @Scheduled(cron = "${scheduled.learnLeaveMsg}")
    public void startJobTimer() {
        if (debug) {
            log.warn("本地环境不启动定时");
            return;
        }
        if (!getLearnLeaveMsgLock()) {
            return;
        }
        log.info("计时离开推送push任务开启！");
        try {
            Date now = new Date();
            Set<String> valStringSet = redisTemplate.opsForZSet().rangeByScore(USERS_LEARN_LEAVE_MSG_SET, now.getTime() - (leaveMinute * 60 * 1000) * 48, (now.getTime() - (leaveMinute * 60 * 1000)));
            if (CollectionUtils.isEmpty(valStringSet)) {
                return;
            }
            log.info("计时离开推送push key = {}", JSON.toJSONString(valStringSet));

            for (String valString : valStringSet) {
                try {

                    List<String> strings = Splitter.on(":").splitToList(valString);
                    Integer userId = Integer.valueOf(strings.get(0));
                    Long startLearnTime = Long.valueOf(strings.get(1));

                    UsersVO users = usersService.findOne(userId.longValue());
                    gyfTaskExecutor.execute(() -> {

                        log.info("推送处理 startLearnTime = {} users = {}", startLearnTime, users);

                        //用户如果已经不是计时状态，push停止
                        if (!users.getStatus().equals(LEARN_TIMING.getCode())) {
                            log.info("用户已经停止学习计时，不推送计时离开push users = {}", users);
                            return;
                        }

                        //如果用户通知消息和当前学习不是同一次，停止push
                        if (Objects.nonNull(users.getStartLearnTime()) && users.getStartLearnTime().getTime() != startLearnTime) {
                            log.info("用户本次计时已经结束  startLearnTime = {} user-leran-time = {} user = {} ", startLearnTime, users.getStartLearnTime().getTime(), users);
                            return;
                        }

                        //如果用户app没有关闭，停止push
                        List<UserOnlineVO> userOnlineList = socketService.getOnline(users.getId()).getData();
                        log.info("userOnlineList = null", userOnlineList);
                        if (Objects.nonNull(userOnlineList) && !userOnlineList.isEmpty() &&
                                (Objects.isNull(userOnlineList.get(0).getTopRun()) || Integer.valueOf(YN.YES.getCode()).equals(userOnlineList.get(0).getTopRun()))) {
                            log.info("用户在线，并且用户APP没有设置在后台，不推送计时离开push users = {} userOnlineList = {}", users, userOnlineList);
                            usersStudyService.addLeaveMsg(LeaveMsgVO.builder().userId(userId).startLearnTime(startLearnTime).uniqueId(UUID.randomUUID().toString()).build());
                            return;
                        }
                        //加10000是为了容错，防止程序执行速度影响计算结果
                        int minute = (int) (System.currentTimeMillis() + 10000 - users.getStartLearnTime().getTime()) / 1000 / 60;
                        String msg = "";
                        if (minute < 60) {
                            msg = minute + "分钟";
                        } else {
                            int hour = minute / 60;
                            msg = hour + "小时";
                            if (hour >= 8) {
                                PushDTO pushMsg = PushDTO.builder().ids(Collections.singletonList(users.getId().toString())).type(PushDTO.TypeEnum.USER.code).content("同桌\n本次学习计时已经结束").build();
                                deskMatePushService.push(pushMsg);
                                return;
                            }
                        }
                        PushDTO pushMsg = PushDTO.builder().ids(Collections.singletonList(users.getId().toString())).type(PushDTO.TypeEnum.USER.code).content("同桌\n本次学习已达" + msg).build();
                        deskMatePushService.push(pushMsg);
                        usersStudyService.addLeaveMsg(LeaveMsgVO.builder().userId(userId).startLearnTime(startLearnTime).uniqueId(UUID.randomUUID().toString()).build());
                        log.info("用户满足推送条件，开始推送计时离开push = {} userOnlineList = {} user = {}", pushMsg, userOnlineList, users);

                    });

                } catch (Exception ex) {
                    log.error("计时离开推送push error，ex = {}", ex);
                }
            }

            redisTemplate.opsForZSet().remove(USERS_LEARN_LEAVE_MSG_SET, valStringSet.toArray(new String[]{}));

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            releaseLearnLeaveMsgLock();
        }
    }


    private boolean getLearnLeaveMsgLock() {

        String key = USERS_LEARN_LEAVE_MSG_LOCK;
        try {
            String result = (String) redisTemplate.execute((RedisCallback<String>) connection -> {
                JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                return commands.set(key, key, "NX", "EX", 50);
            });
            return !org.apache.commons.lang3.StringUtils.isEmpty(result);
        } catch (Exception e) {
            log.error("getLearnLeaveMsgLock error", e);
        }
        return false;
    }

    private void releaseLearnLeaveMsgLock() {
        String key = USERS_LEARN_LEAVE_MSG_LOCK;
        try {
            redisTemplate.delete(key);
        } catch (Exception e) {
            log.error("releaseLearnLeaveMsgLock error", e);
        }
    }

    private boolean getRefreshUserOnlineLock() {

        String key = REFRESH_USER_ONLINE_STATE;
        try {
            String result = (String) redisTemplate.execute((RedisCallback<String>) connection -> {
                JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                return commands.set(key, key, "NX", "EX", 50);
            });
            return !org.apache.commons.lang3.StringUtils.isEmpty(result);
        } catch (Exception e) {
            log.error("getLearnLeaveMsgLock error", e);
        }
        return false;
    }

    private void releaseRefreshUserOnlineLock() {
        String key = REFRESH_USER_ONLINE_STATE;
        try {
            redisTemplate.delete(key);
        } catch (Exception e) {
            log.error("releaseLearnLeaveMsgLock error", e);
        }
    }

}
